package org.fastsyncer.manager.increment.ldap;

import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.LinkedBlockingQueue;

import javax.naming.Binding;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingEnumeration;
import javax.naming.NamingException;
import javax.naming.directory.Attribute;
import javax.naming.directory.Attributes;
import javax.naming.directory.SearchResult;
import javax.naming.event.EventContext;
import javax.naming.event.NamespaceChangeListener;
import javax.naming.event.NamingEvent;
import javax.naming.event.NamingExceptionEvent;
import javax.naming.event.NamingListener;
import javax.naming.event.ObjectChangeListener;

import org.apache.commons.lang.StringUtils;
import org.fastsyncer.common.constant.CommonConstant;
import org.fastsyncer.common.constant.ConnectorConstant;
import org.fastsyncer.common.util.ApplicationUtil;
import org.fastsyncer.connector.common.LdapConstant;
import org.fastsyncer.manager.store.DataCenter;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * LdapListener 事件监听器
 * 
 * @ClassName: LdapListener
 * @Description: 监听ldap新增、修改、刪除操作 <br />
 *               执行流程：
 *               <ul>
 *               <li>注册LDAP监听</li>
 *               <li>将增量日志分类放入CHANELL队列</li>
 *               <li>每次请求都从CHANELL拉取增量日志</li>
 *               </ul>
 * @author: YangPei
 * @date: 2017年8月22日 下午6:39:05
 */
public class LdapListener {

    private static Logger logger = LoggerFactory.getLogger(LdapListener.class);

    private static LdapListener instance = new LdapListener();

    private LdapListener() {
    }

    public static LdapListener getInstance() {
        return instance;
    }

    // 并发大小
    private final static int CAPACITY = 1000;

    // 创建监听器，新增、删除、重命名属性
    private final static String NAMESPACE_LISTENER = "NamespaceChangeListener";

    // 修改
    private final static String OBJECT_LISTENER = "ObjectChangeListener";

    // 监听器集合 NamespaceChangeListener和ObjectChangeListener
    private final static Map<String, NamingListener> MAP_LISTENER = new HashMap<String, NamingListener>(2);

    // 注册的驱动监听
    private final static Map<String, EventContext> RUNNING = new HashMap<String, EventContext>();

    // 监听到的增量日志,初始化容器大小为3000
    private final static Map<String, LinkedBlockingQueue<JSONObject>> CHANELL = new HashMap<String, LinkedBlockingQueue<JSONObject>>();

    /**
     * 获取增量对应的驱动名称
     * 
     * @Title: getDriverName
     * @Description: 获取增量对应的驱动名称
     * @param url
     *            监听地址
     * @param nameInNamespace
     *            监听的域
     * @throws NamingException
     * @return: String
     */
    private static String getDriverName(String url, String nameInNamespace) throws NamingException {
        String driverName = null;
        for (Map.Entry<String, EventContext> driver : RUNNING.entrySet()) {
            EventContext context = driver.getValue();
            Hashtable<?, ?> environment = context.getEnvironment();
            String providerUrl = (String) environment.get(Context.PROVIDER_URL);
            String space = context.getNameInNamespace();

            // 如果监听的地址和域一致,那么当前的增量就是该驱动的数据
            if (StringUtils.equals(providerUrl, url) && StringUtils.equals(space, nameInNamespace)) {
                driverName = driver.getKey();
                break;
            }
        }
        return driverName;
    }

    /**
     * 检查消息类型是否为SearchResult，否则抛出异常
     * 
     * @Title: validateBinding
     * @Description: 检查消息类型是否为SearchResult，否则抛出异常
     * @param binding
     *            消息
     * @throws IllegalArgumentException
     * @return: SearchResult
     */
    private static SearchResult validateBinding(Binding binding) throws IllegalArgumentException {
        if (!(binding instanceof SearchResult)) {
            throw new IllegalArgumentException("Parameter must be an instance of SearchResult");
        }
        return (SearchResult) binding;
    }

    /**
     * 解析增量属性
     * 
     * @Title: parseSearchResult
     * @Description: 解析增量属性
     * @param searchResult
     *            增量属性
     * @param data
     *            要组装进的数组
     * @throws IllegalArgumentException
     * @throws NamingException
     * @throws JSONException
     * @return: void
     */
    private static void parseSearchResult(SearchResult searchResult, JSONArray data) throws IllegalArgumentException, NamingException, JSONException {
        // 属性
        Attributes attributes = searchResult.getAttributes();
        if (attributes == null) {
            throw new IllegalArgumentException("The properties of the sync are empty!");
        }

        // 获取属性名
        NamingEnumeration<String> iDs = attributes.getIDs();
        while (iDs.hasMoreElements()) {
            String attrName = (String) iDs.nextElement();
            // 在LDAP中，同属性名可能有多个，所以获取的时候应该用集合
            Attribute atts = attributes.get(attrName);
            int size = atts.size();
            // 排除无效的属性
            if (size < 1) {
                continue;
            }
            // 遍历属性
            JSONArray list = new JSONArray();
            for (int i = 0; i < size; i++) {
                list.put(atts.get(i));
            }
            // 组装属性名和值
            JSONObject attr = new JSONObject();
            attr.put("name", attrName);
            attr.put("value", list);
            data.put(attr);
        }
    }

    /**
     * 解析增量事件
     * 
     * @Title: parseLog
     * @Description: 解析增量事件，将解析后的日志发送到消息队列中
     * @param opr
     *            事件类型
     * @param evt
     *            监听的事件包装（包含监听的服务器配置、变化前后的数据）
     * @return: void
     */
    private static void parseLog(String opr, NamingEvent evt) {
        try {
            // 1、排除非新增、修改和删除操作
            if (!StringUtils.equals(ConnectorConstant.OPERTION_INSERT, opr) && !StringUtils.equals(ConnectorConstant.OPERTION_UPDATE, opr)
                    && !StringUtils.equals(ConnectorConstant.OPERTION_DELETE, opr)) {
                return;
            }

            // 2、获取上下文
            EventContext eventContext = evt.getEventContext();
            // 监听的地址
            Hashtable<?, ?> environment = eventContext.getEnvironment();
            // 找到对应的驱动名称
            String taskName = getDriverName((String) environment.get(Context.PROVIDER_URL), eventContext.getNameInNamespace());
            // 如果没有匹配注册监听的驱动，则直接跳过此流程
            if (StringUtils.isBlank(taskName)) {
                return;
            }

            // 3、检查是否已创建队列
            if (CHANELL.get(taskName) == null) {
                CHANELL.put(taskName, new LinkedBlockingQueue<JSONObject>(CAPACITY));
            }
            LinkedBlockingQueue<JSONObject> queue = CHANELL.get(taskName);

            // 4、封装增量消息格式
            JSONObject handle = new JSONObject();
            handle.put("eventType", opr);
            handle.put("before", new JSONArray());
            handle.put("after", new JSONArray());

            // 5、根据不同的操作事件，包装不同的数据格式
            switch (opr) {
            case ConnectorConstant.OPERTION_INSERT:
                // 获取变化后的数据
                Binding bindingInsert = evt.getNewBinding();
                // UID
                handle.put("base", bindingInsert.getName());
                // 检查消息类型是否为SearchResult，否则抛出异常
                SearchResult searchResultInsert = validateBinding(bindingInsert);
                // 获取属性
                parseSearchResult(searchResultInsert, handle.getJSONArray("after"));
                break;
            case ConnectorConstant.OPERTION_UPDATE:
                // 获取变化后的数据
                Binding bindingUpdate = evt.getNewBinding();
                handle.put("base", bindingUpdate.getName());
                SearchResult searchResultUpdate = validateBinding(bindingUpdate);
                // 获取属性
                parseSearchResult(searchResultUpdate, handle.getJSONArray("after"));
                break;
            case ConnectorConstant.OPERTION_DELETE:
                // 获取变化前的数据
                Binding bindingDelete = evt.getOldBinding();
                handle.put("base", bindingDelete.getName());
                SearchResult searchResultDelete = validateBinding(bindingDelete);
                // 获取属性
                parseSearchResult(searchResultDelete, handle.getJSONArray("before"));
                break;
            default:
                break;
            }
            // 6、最后发送到消息队列（将元素插入队列，成功返回true，如果当前没有可用的空间，则返回false）
            queue.offer(handle);
        } catch (NamingException e) {
            logger.error(e.getMessage());
        } catch (JSONException e) {
            logger.error(e.getMessage());
        }
    }

    /**
     * 创建监听器上下文
     * 
     * @Title: create
     * @Description: 创建监听器上下文
     * @param baseDN
     * @param url
     * @param principal
     * @param credentials
     * @throws NamingException
     * @return: EventContext
     */
    private static EventContext createContext(String baseDN, String url, String principal, String credentials, Integer subtreeScope) throws NamingException {
        // 注册监听的域
        // 配置上下文
        Hashtable<String, Object> env = new Hashtable<String, Object>();
        env.put(Context.INITIAL_CONTEXT_FACTORY, "com.sun.jndi.ldap.LdapCtxFactory");
        env.put(Context.PROVIDER_URL, url);
        env.put(Context.SECURITY_PRINCIPAL, principal);
        env.put(Context.SECURITY_CREDENTIALS, credentials);
        EventContext ctx = (EventContext) (new InitialContext(env).lookup(baseDN));

        // 注册监听器
        NamingListener namingListener = MAP_LISTENER.get(NAMESPACE_LISTENER);
        NamingListener objectListener = MAP_LISTENER.get(OBJECT_LISTENER);
        ctx.addNamingListener("", subtreeScope, namingListener);
        ctx.addNamingListener("", subtreeScope, objectListener);
        return ctx;
    }

    static {
        // 创建监听器，新增、删除、重命名属性
        NamingListener namespaceListener = new NamespaceChangeListener() {
            // 新增
            public void objectAdded(NamingEvent evt) {
                parseLog(ConnectorConstant.OPERTION_INSERT, evt);
            }

            // 删除
            public void objectRemoved(NamingEvent evt) {
                parseLog(ConnectorConstant.OPERTION_DELETE, evt);
            }

            // 重命名域名称 template.rename("uid=zhangsan", "uid=zhangsan666");
            public void objectRenamed(NamingEvent evt) {
            }

            // 异常监听
            public void namingExceptionThrown(NamingExceptionEvent evt) {
                logger.error(evt.getException().getMessage());
            }
        };

        // 修改
        NamingListener objectListener = new ObjectChangeListener() {
            // 更新
            public void objectChanged(NamingEvent evt) {
                parseLog(ConnectorConstant.OPERTION_UPDATE, evt);
            }

            // 异常监听
            public void namingExceptionThrown(NamingExceptionEvent evt) {
                logger.error("异常监听");
            }
        };

        MAP_LISTENER.put(NAMESPACE_LISTENER, namespaceListener);
        MAP_LISTENER.put(OBJECT_LISTENER, objectListener);
    }
    
    /**
     * 拉取LDAP增量日志
     * @Title: pull 
     * @Description: 拉取LDAP增量日志
     * @param params 驱动配置
     * @return: JSONArray 增量日志
     * @throws NamingException 
     */
    public JSONArray pull(Map<String, String> params) throws NamingException {
        // 驱动名称
        String taskId = params.get("taskId");
        EventContext ctx = RUNNING.get(taskId);

        // 检查是否已注册过驱动监听器
        if (ctx == null) {
            // 创建监听任务
            this.addTask(taskId, params);
        }

        // 检查驱动是否发生变更
        String key = CommonConstant.DRIVER_CONFIG_UPDATE + taskId;
        String update = ApplicationUtil.getKey(key);
        if (StringUtils.isNotBlank(update) && StringUtils.equals(update, "true")) {
            // 移除该任务
            this.removeTask(taskId);
            // 创建监听任务
            this.addTask(taskId, params);
            // 移除标示
            ApplicationUtil.setKey(key, "");
        }

        // 获取指定驱动的增量日志
        LinkedBlockingQueue<JSONObject> queue = CHANELL.get(taskId);
        if (queue == null || queue.isEmpty()) {
            return null;
        }
        // 返回的数据
        JSONArray msg = new JSONArray();
        while (!queue.isEmpty()) {
            msg.put(queue.poll());
        }
        return msg;
    }

    private void removeTask(String taskId) throws NamingException {
        // 注销监听器
        EventContext con = RUNNING.get(taskId);
        if (null != con) {
            con.close();
            RUNNING.remove(taskId);
        }
    }

    private void addTask(String taskId, Map<String, String> params) throws NamingException {
        // 默认监听器
        String baseDN = params.get("base");
        String url = params.get("url");
        String principal = params.get("userDn");
        String credentials = params.get("password");
        String scope = params.get("searchScope");
        Integer searchScope = LdapConstant.OPERS.get(scope);
        EventContext context = createContext(baseDN, url, principal, credentials, searchScope);
        if (null != context) {
            RUNNING.put(taskId, context);
        }
    }
    
    public void start(){
        new LdapTaskThread().start();
    }
    
    class LdapTaskThread extends Thread {

        LdapTaskThread() {
            setName("fastsyncer ldap task thread");
        }
        
        public void run() {
            while(true){
                // 刷新监听任务
                for (Entry<String, EventContext> t : RUNNING.entrySet()) {
                    // 如果任务已被删除
                    if (null == DataCenter.getInstance().getMapping(t.getKey())) {
                        try {
                            // 移除该任务
                            removeTask(t.getKey());
                        } catch (NamingException e) {
                            logger.error(e.getClass() + "\t" + e.getMessage() + "\t" + e.getStackTrace());
                        }
                    }
                }
                
                //扫描任务间隔时间
                try {
                    Thread.sleep(30000);
                } catch (InterruptedException e) {
                }
            }
        }

    }

}